package com.lengxf.flink.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//用 8没问题
//17执行时需要增加执行参数
//        --add-opens java.base/java.lang=ALL-UNNAMED
//        --add-opens java.base/java.util=ALL-UNNAMED
//        --add-opens java.base/java.nio=ALL-UNNAMED
//        --add-opens java.base/sun.nio.ch=ALL-UNNAMED

//流处理读取文件
//文件为有界数据流
public class WordCountStreamDemo {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.读取数据
        DataStreamSource<String> lineDS = env.readTextFile("flink/src/main/resources/word.txt");
        //3.处理数据
        SingleOutputStreamOperator<Tuple2<String, Integer>> singleOutputStreamOperator = lineDS.flatMap(getStringTuple2FlatMapFunction())
                .returns(Types.TUPLE(Types.STRING, Types.INT));
        //分组
        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = singleOutputStreamOperator.keyBy(v -> v.f0);
        //聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum(1);
        //4.输出数据
        sum.print();
        //5.执行
        env.execute();
    }

    private static FlatMapFunction<String, Tuple2<String, Integer>> getStringTuple2FlatMapFunction() {
        return (s, collector) -> {
            String[] words = s.split(" ");
            for (String word : words) {
                collector.collect(Tuple2.of(word, 1));
            }
        };
    }

}
